/*
 * Project Name spring-boot-tools
 * File Name RabbitmqProperties
 * Package Name com.huxiaosu.tools.mq.config
 * Create Time 2024/7/9
 * Create by name：liujie
 */
package com.huxiaosu.tools.mq.config;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Component;

/**
 * Description
 *
 * @author liujie
 * @date 2024/7/9 10:50
 */
@Slf4j
@Configuration
public class RabbitmqConfiguration implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback{
    @Autowired
    private RabbitProperties rabbitProperties;
    @Autowired
    private RabbitmqProperties rabbitmqProperties;

     private final static String DEFAULT_EXCHANGE_TYPE = "direct";
    @Bean
    @Primary
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setHost(rabbitProperties.getHost());
        cachingConnectionFactory.setPort(rabbitProperties.getPort());
        cachingConnectionFactory.setUsername(rabbitProperties.getUsername());
        cachingConnectionFactory.setPassword(rabbitProperties.getPassword());
        cachingConnectionFactory.setVirtualHost(rabbitProperties.getVirtualHost());
        return cachingConnectionFactory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
        //消息被（broker）接收的回调
        rabbitTemplate.setConfirmCallback(this);
        //消息被（broker）退回的回調
        rabbitTemplate.setReturnsCallback(this);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        return rabbitTemplate;
    }
    @Bean
    public SimpleRabbitListenerContainerFactory listenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        /* setConnectionFactory：设置spring-amqp的ConnectionFactory。 */
        factory.setConnectionFactory(connectionFactory());
        /* 消息序列化类型 */
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        /* setConcurrentConsumers：设置每个MessageListenerContainer将会创建的Consumer的最小数量，默认是1个。 */
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        /* setPrefetchCount：设置每次请求发送给每个Consumer的消息数量。 */
        factory.setPrefetchCount(1);
        /* 是否设置Channel的事务。 */
        factory.setChannelTransacted(false);
        /* 设置当rabbitmq收到nack/reject确认信息时的处理方式，设为true，扔回queue头部，设为false，丢弃。 */
        factory.setDefaultRequeueRejected(true);

        factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
        return (factory);
    }

    @Bean(name = "${tools.rabbitmq.exchange:tools.demo}")
    public Exchange exchange() {
        if(DEFAULT_EXCHANGE_TYPE.equalsIgnoreCase(rabbitmqProperties.getExchangeType())){
            log.info("< create direct exchange bean [{}]>",rabbitmqProperties.getExchange());
            return new DirectExchange(rabbitmqProperties.getExchange(), true, false);
        }
        log.info("< create fanout exchange [{}]>",rabbitmqProperties.getExchange());
        return new FanoutExchange(rabbitmqProperties.getExchange(), true, false);

    }
    @Bean(name = "${tools.rabbitmq.request:tools.demo.request}")
    public Queue requestQueue() {
        log.info("< create request queue [{}]>",rabbitmqProperties.getRequest());
        return new Queue(rabbitmqProperties.getRequest(), true, false, false);
    }
    @Bean(name = "${tools.rabbitmq.response:tools.demo.response}")
    public Queue responseQueue() {
        log.info("< create response queue [{}]>",rabbitmqProperties.getResponse());
        return new Queue(rabbitmqProperties.getResponse(), true, false, false);
    }

    @Bean
    public Binding bindRequest(){
        log.info("< bind request queue [{}] with routingKey = [{}]>",rabbitmqProperties.getRequest(),rabbitmqProperties.getRequestKey());
        return BindingBuilder.bind(requestQueue()).to(exchange()).with(rabbitmqProperties.getRequestKey()).noargs();
    }
    @Bean
    public Binding bindResponse(){
        log.info("< bind response queue [{}] with routingKey = [{}]>",rabbitmqProperties.getResponse(),rabbitmqProperties.getResponseKey());
        return BindingBuilder.bind(responseQueue()).to(exchange()).with(rabbitmqProperties.getResponseKey()).noargs();
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.debug("消息唯一标识：" + correlationData);
        log.debug("确认结果：" + ack);
        log.debug("失败原因：" + cause);
    }

    @Override
    public void returnedMessage(ReturnedMessage returned) {
        log.debug("消息主体 message : " + returned.getMessage());
        log.debug("消息主体 replyCode : " + returned.getReplyCode());
        log.debug("描述：" + returned.getReplyText());
        log.debug("消息使用的交换器 exchange : " + returned.getExchange());
        log.debug("消息使用的路由键 routing : " + returned.getRoutingKey());
    }
}